package job;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties.JobPropertiesEnum;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 替换 frxsframework-elasticjob-1.0.8.jar，elastic-job启动问题解决
 * com.frxs.framework.elasticjob.parser.ElasticJobConfParser
 *
 * @author MingWu.Chen
 * @date 2022年06月13日 19:39
 */
public class ElasticJobConfParser implements ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(ElasticJobConfParser.class);
    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;
    @Autowired
    private JobZookeeperProperties jobZookeeperProperties;

	@Autowired
	private ElasticJobConfWhitelistHandler elasticJobConfWhitelistHandler;

	@Autowired
	private Environment environment;

	@Autowired
	private ElasticJobConfWhitelistProperties elasticJobConfWhitelistProperties;

    public ElasticJobConfParser() {
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		try {
            Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(ElasticJobConfig.class);
            Iterator var3 = beanMap.values().iterator();

            while(var3.hasNext()) {
                Object confBean = var3.next();
                Class<?> clz = confBean.getClass();
                String jobTypeName;
                if (clz.getName().indexOf("$") > 0) {
                    jobTypeName = clz.getName();
                    clz = Class.forName(jobTypeName.substring(0, jobTypeName.indexOf("$")));
                }
				if (clz.getInterfaces().length > 0) {
					jobTypeName = clz.getInterfaces()[0].getSimpleName();
				} else {
					jobTypeName = ElasticJobTypeEnum.SIMPLE.getType();
				}
				ElasticJobConfig conf = clz.getAnnotation(ElasticJobConfig.class);
                String jobClass = clz.getName();
                String jobName = this.jobZookeeperProperties.getNamespace() + "." + conf.name();
                String cron = conf.cron();
                String shardingItemParameters = conf.shardingItemParameters();
                String description = conf.description();
                String jobParameter = conf.jobParameter();
                String jobExceptionHandler = conf.jobExceptionHandler();
                String executorServiceHandler = conf.executorServiceHandler();
                String jobShardingStrategyClass = conf.jobShardingStrategyClass();
                String eventTraceRdbDataSource = conf.eventTraceRdbDataSource();
                String scriptCommandLine = conf.scriptCommandLine();
                boolean failover = conf.failover();
                boolean misfire = conf.misfire();
                boolean overwrite = conf.overwrite();
				boolean disabled = conf.disabled();

				//如果是默认值false,则读取一下全局自定义配置
				if (!disabled) {
					disabled = elasticJobConfWhitelistHandler.isDisable(jobClass);
					jobName = warpJobName(jobName, disabled);
				}
				boolean monitorExecution = conf.monitorExecution();
				boolean streamingProcess = conf.streamingProcess();
                int shardingTotalCount = conf.shardingTotalCount();
                int monitorPort = conf.monitorPort();
                int maxTimeDiffSeconds = conf.maxTimeDiffSeconds();
                int reconcileIntervalMinutes = conf.reconcileIntervalMinutes();
                JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(jobName, cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).description(description).failover(failover).jobParameter(jobParameter).misfire(misfire).jobProperties(JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), jobExceptionHandler).jobProperties(JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), executorServiceHandler).build();
                JobTypeConfiguration typeConfig = null;
                if (ElasticJobTypeEnum.SIMPLE.getType().equals(jobTypeName)) {
                    typeConfig = new SimpleJobConfiguration(coreConfig, jobClass);
                }

                if (ElasticJobTypeEnum.DATAFLOW.getType().equals(jobTypeName)) {
                    typeConfig = new DataflowJobConfiguration(coreConfig, jobClass, streamingProcess);
                }

                if (ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                    typeConfig = new ScriptJobConfiguration(coreConfig, scriptCommandLine);
                }

                LiteJobConfiguration jobConfig = LiteJobConfiguration.newBuilder((JobTypeConfiguration)typeConfig).overwrite(overwrite).disabled(disabled).monitorPort(monitorPort).monitorExecution(monitorExecution).maxTimeDiffSeconds(maxTimeDiffSeconds).jobShardingStrategyClass(jobShardingStrategyClass).reconcileIntervalMinutes(reconcileIntervalMinutes).build();
                List<BeanDefinition> elasticJobListeners = this.getTargetElasticJobListeners(conf);
                BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
                //factory.setInitMethodName("init");
                factory.setScope("prototype");
                if (!ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                    factory.addConstructorArgValue(confBean);
                }

                factory.addConstructorArgValue(this.zookeeperRegistryCenter);
                factory.addConstructorArgValue(jobConfig);
                if (StringUtils.hasText(eventTraceRdbDataSource)) {
                    BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
                    rdbFactory.addConstructorArgReference(eventTraceRdbDataSource);
                    factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
                }

                factory.addConstructorArgValue(elasticJobListeners);
                DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory)applicationContext.getAutowireCapableBeanFactory();
                String registerBeanName = conf.name() + "SpringJobScheduler";
                defaultListableBeanFactory.registerBeanDefinition(registerBeanName, factory.getBeanDefinition());
                SpringJobScheduler springJobScheduler = (SpringJobScheduler)applicationContext.getBean(registerBeanName);
                springJobScheduler.init();
                log.info("启动elastic-job的作业：" + jobName);
            }

            log.info("共计启动elastic-job的作业{}个", beanMap.values().size());
        } catch (Exception var37) {
            log.error("elasticJob启动异常，进行强制退出", var37);
            System.exit(1);
        }

    }

	/**
	 * 自己include进来的job的后缀名称
	 * @param jobName
	 * @param disabled
	 * @return
	 */
	private String warpJobName(String jobName, boolean disabled) {
		if (!disabled) {
			boolean hasCustomInclude = elasticJobConfWhitelistHandler.isHasCustomInclude();
			if (hasCustomInclude) {
				String includeJobNameSuffix = elasticJobConfWhitelistHandler.getIncludeJobNameSuffix();
				if (null != includeJobNameSuffix && includeJobNameSuffix.trim().length() > 0) {
					jobName = jobName +"_"+ includeJobNameSuffix.trim();
				}
			}
		}
		return jobName;
	}

	private List<BeanDefinition> getTargetElasticJobListeners(ElasticJobConfig conf) {
        List<BeanDefinition> result = new ManagedList(2);
        String listeners = conf.listener();
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope("prototype");
            result.add(factory.getBeanDefinition());
        }

        String distributedListeners = conf.distributedListener();
        long startedTimeoutMilliseconds = conf.startedTimeoutMilliseconds();
        long completedTimeoutMilliseconds = conf.completedTimeoutMilliseconds();
        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope("prototype");
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }

        return result;
    }
}